/*
 * Copyright (c) 2016, www.zm0618.com All Rights Reserved..
 */

package com.ibetter;

import com.ibetter.mr.grade.medic.browse.GradeUserBrowseMapper;
import com.ibetter.mr.grade.medic.category.GradeUserCatScoreMapper;
import com.ibetter.mr.grade.medic.category.GradeUserCatScoreMergeMapper;
import com.ibetter.mr.grade.medic.category.GradeUserCatScoreMergeReducer;
import com.ibetter.mr.grade.medic.category.GradeUserCatScoreReducer;
import com.ibetter.mr.grade.medic.collect.GradeUserCollectMapper;
import com.ibetter.mr.grade.medic.comment.GradeUserCommentMapper;
import com.ibetter.mr.grade.medic.common.GradeCommonReducer;
import com.ibetter.mr.grade.medic.score.GradeScoreCounterMapper;
import com.ibetter.mr.grade.medic.score.GradeScoreCounterReducer;
import com.ibetter.mr.grade.medic.share.GradeUserShareMapper;
import com.ibetter.mr.grade.medic.tag.GradeUserTagScoreMapper;
import com.ibetter.mr.grade.medic.tag.GradeUserTagScoreMergeMapper;
import com.ibetter.mr.grade.medic.tag.GradeUserTagScoreMergeReducer;
import com.ibetter.mr.grade.medic.tag.GradeUserTagScoreReducer;
import com.ibetter.mr.grade.medic.voteup.GradeUserVoteMedicMapper;
import com.ibetter.mr.grade.medic.voteup.GradeUserVoteMedicReducer;
import com.ibetter.mr.parse.medic.CleanDumpShareMapper;
import com.ibetter.mr.parse.medic.CleanDumpShareReducer;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;

/**
 * Created by zhaojun on 2016/12/06.
 */
public class App {

    public static void main(String[] args) {
       /* Configuration config=new Configuration();
        try {
            Job job = Job.getInstance(config);
            job.setJarByClass(App.class);
            String jobName=args[0];
            if (StringUtils.isNotBlank(jobName)){

                if (StringUtils.equals(jobName,"voteup")){//点赞
                    userVoteMedicJob(args,  job);
                }
                if(StringUtils.equals(jobName,"share")){//分享
                    userShareMedicJob(args,job);
                }
                if (StringUtils.equals(jobName,"comment")){//评论
                    userCommentMedicJob(args,job);
                }

                if (StringUtils.equals(jobName,"collect")){//收藏
                    userCollectMedicJob(args,job);
                }
                if (StringUtils.equals(jobName,"browse")){//浏览
                    userBrowseMedicJob(args,job);
                }

                if (StringUtils.equals(jobName,"scorecount")){//计算这分数
                    userScoreCount(args,job);
                }
                if(StringUtils.equals(jobName,"cleanRepeatedShare")){//去分享去重复
                    cleanDumpShare(args,  job);
                }

                if (StringUtils.equals(jobName,"catscore")){//整理分类打分
                    userCatScoreJob(args,  job);
                }

                if(StringUtils.equals(jobName,"catscoremerge")){//合并打分
                    userCatScoreMerge(args,  job);
                }
                if (StringUtils.equals(jobName,"tagscore")){//整理TAG打分
                    userTagScoreJob(args,  job);
                }

                if(StringUtils.equals(jobName,"tagscoremerge")){//合并打分
                    userTagScoreMerge(args,  job);
                }

                job.waitForCompletion(true);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }*/

        /*try {
            InputStream stream = App.class.getClassLoader().getResourceAsStream("part-r-00000");
            LineIterator iterator = IOUtils.lineIterator(stream, "UTF-8");
            while (iterator.hasNext()){
                 String line=iterator.nextLine();
                 String[] s = StringUtils.split(line, "\t");
                System.out.println(s[0]);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }*/

        testReadHdfs();
    }

    private static void userTagScoreMerge(String[] args, Job job) throws IOException {
        job.setJobName("userTagScoreMerge");
        job.setMapperClass(GradeUserTagScoreMergeMapper.class);
        job.setReducerClass(GradeUserTagScoreMergeReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FloatWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void userTagScoreJob(String[] args, Job job) throws IOException {

        job.setJobName("userTagScoreJob");
        job.setMapperClass(GradeUserTagScoreMapper.class);
        job.setReducerClass(GradeUserTagScoreReducer.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        String[] inputs=new String[args.length-2];
        for (int i=1;i<args.length-1;i++){
            inputs[i-1]=args[i];
        }
        FileInputFormat.addInputPaths(job,StringUtils.join(inputs,","));
        FileOutputFormat.setOutputPath(job,new Path(args[args.length-1]));
    }

    private static void userCatScoreMerge(String[] args, Job job) throws IOException {

        job.setJobName("userCatScoreMergeJob");
        job.setMapperClass(GradeUserCatScoreMergeMapper.class);
        job.setReducerClass(GradeUserCatScoreMergeReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FloatWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void userCatScoreJob(String[] args, Job job) throws IOException {
        job.setJobName("userCatScoreJob");
        job.setMapperClass(GradeUserCatScoreMapper.class);
        job.setReducerClass(GradeUserCatScoreReducer.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        String[] inputs=new String[args.length-2];
        for (int i=1;i<args.length-1;i++){
            inputs[i-1]=args[i];
        }
        FileInputFormat.addInputPaths(job,StringUtils.join(inputs,","));
        FileOutputFormat.setOutputPath(job,new Path(args[args.length-1]));

    }

    private static void userScoreCount(String[] args, Job job) throws IOException {
        job.setJobName("userScoreCount");
        job.setMapperClass(GradeScoreCounterMapper.class);
        job.setReducerClass(GradeScoreCounterReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        String[] inputs=new String[args.length-2];
        for (int i=1;i<args.length-1;i++){
            inputs[i-1]=args[i];
        }
        FileInputFormat.addInputPaths(job,StringUtils.join(inputs,","));
        FileOutputFormat.setOutputPath(job,new Path(args[args.length-1]));
    }

    private static void userBrowseMedicJob(String[] args, Job job) throws IOException {
        job.setJobName("userBrowseMedicJob");
        job.setMapperClass(GradeUserBrowseMapper.class);
        job.setReducerClass(GradeCommonReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));

    }

    private static void userCollectMedicJob(String[] args, Job job) throws IOException {
        job.setJobName("userCollectMedicJob");
        job.setMapperClass(GradeUserCollectMapper.class);
        job.setReducerClass(GradeCommonReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));

    }

    private static void userCommentMedicJob(String[] args, Job job) throws IOException {
        job.setJobName("userCommentMedicJob");
        job.setMapperClass(GradeUserCommentMapper.class);
        job.setReducerClass(GradeCommonReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void userShareMedicJob(String[] args, Job job) throws IOException {
        System.out.println(args[1]);
        System.out.println(args[2]);
        job.setJobName("userShareMedicJob");
        job.setMapperClass(GradeUserShareMapper.class);
        job.setReducerClass(GradeCommonReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void userVoteMedicJob(String[] args, Job job) throws IOException {
        job.setJobName("userVoteMedicJob");
        job.setMapperClass(GradeUserVoteMedicMapper.class);
        job.setReducerClass(GradeUserVoteMedicReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void cleanDumpShare(String[] args, Job job) throws IOException {
        job.setJobName("cleanDumpSharedJob");
        job.setMapperClass(CleanDumpShareMapper.class);
        job.setReducerClass(CleanDumpShareReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[1]));
        FileOutputFormat.setOutputPath(job,new Path(args[2]));
    }

    private static void testReadHdfs() {
        try {
            Configuration conf=new Configuration();
            URI uri=new URI("hdfs://192.168.10.21:9000");
            FileSystem fileSystem = FileSystem.get(uri, conf);
           /* RemoteIterator<LocatedFileStatus> remoteIter = fileSystem.listFiles(new Path("/user/hive/warehouse/bi_zm_db.db/"), true);
            while(remoteIter.hasNext()){
                LocatedFileStatus filestatus = remoteIter.next();
                System.out.println(filestatus.getPath());
            }
*/
            Path path=new Path("/user/hive/warehouse/bi_zm_db.db/zm_coraic_conference_member/zm_coraic_conference_member.txt");
            FileStatus filestatus = fileSystem.getFileStatus(path);
            short replication = filestatus.getReplication();
            System.out.println(replication);
            System.out.println(filestatus.getBlockSize());
            BlockLocation[] locations = fileSystem.getFileBlockLocations(filestatus, 0, 128 * 1024 * 1024*2);

            for (int i = 0; i <locations.length ; i++) {
                String[] topologyPaths = locations[i].getTopologyPaths();
                for (int j = 0; j <topologyPaths.length ; j++) {
                    System.out.println(topologyPaths[j]);
                }
                System.out.println(locations[i].getTopologyPaths());
            }
           /* FSDataInputStream dis = fileSystem.open(path);
            int l = dis.available();
            ByteBuffer buffer = ByteBuffer.allocate(l);
            dis.read(buffer);
            buffer.flip();
            System.out.println( new String(buffer.array(),"utf-8"));*/
        } catch (URISyntaxException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
